package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * Author: Felix
 * Date: 2021/9/15
 * Desc:  专门用于进行业务数据动态分流的类
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private OutputTag<JSONObject> dimTag;

    private MapStateDescriptor<String, TableProcess> mapStateDes;

    private Connection conn;

    public TableProcessFunction(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDes) {
        this.dimTag = dimTag;
        this.mapStateDes = mapStateDes;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        //建立连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //处理主流中的数据  获取状态-->从状态中获取配置--->根据配置进行分流
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDes);

        //获取业务操作的表名
        String tableName = jsonObj.getString("table");

        //获取业务数据的操作类型
        String type = jsonObj.getString("type");

        //注意：如果使用maxwell的bootstrap处理历史数据，操作类型是bootstrap-insert ,需要统一一下
        if (type.equals("bootstrap-insert")) {
            type = "insert";
            jsonObj.put("type", type);
        }
        //拼接获取配置的key
        String key = tableName + ":" + type;
        //从状态中获取配置
        TableProcess tableProcess = broadcastState.get(key);

        if (tableProcess != null) {
            //不管是事实还是维度，都需要将输出目的地 添加到json对象中
            jsonObj.put("sink_table",tableProcess.getSinkTable());

            //对字段进行过滤
            filterColumn(jsonObj.getJSONObject("data"),tableProcess.getSinkColumns());

            //在配置表中，有对应的配置  继续判断是维度数据还是事实数据
            String sinkType = tableProcess.getSinkType();
            if(TableProcess.SINK_TYPE_HBASE.equals(sinkType)){
                //维度数据  输出到维度侧输出流
                ctx.output(dimTag,jsonObj);
            }else if(TableProcess.SINK_TYPE_KAFKA.equals(sinkType)){
                //事实数据  输出到主流
                out.collect(jsonObj);
            }
        }else{
            System.out.println("No this key in TableProcess:" +key);
        }
    }

    //字段过滤  data:{"id":10,"tm_name":"atguigu","log_url":"xxx"}  sink_column :id,tm_name
    private void filterColumn(JSONObject dataJsonObj, String fieldStr) {
        //将保留的字段进行分割
        String[] fieldArr = fieldStr.split(",");
        //为了判断包含方便，将数组转换为集合
        List<String> fieldList = Arrays.asList(fieldArr);

        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        /*Iterator<Map.Entry<String, Object>> it = entrySet.iterator();
        for (;it.hasNext();) {
            Map.Entry<String, Object> entry = it.next();
            if(!fieldList.contains(entry.getKey())){
                it.remove();
            }
        }*/
        entrySet.removeIf(entry->!fieldList.contains(entry.getKey()));


    }

    //处理广播流中的数据 读取配置-->放到状态中
    //{"database":"gmall0408_realtime","data":{"sourceTable":"aaa","opxx":33,"xx":33},"type":"insert","table":"table_process"}
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //将json字符串转换成json对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //通过json对象获取其中的data属性,并转换为TableProcess对象
        TableProcess tableProcess = JSON.parseObject(jsonObj.getString("data"), TableProcess.class);
        //获取业务表名称
        String sourceTable = tableProcess.getSourceTable();
        //获取配置的操作类型
        String operateType = tableProcess.getOperateType();
        //获取输出类型           hbase---维度       kafka---事实
        String sinkType = tableProcess.getSinkType();
        //输出的目的地
        String sinkTable = tableProcess.getSinkTable();
        //建表主键
        String sinkPk = tableProcess.getSinkPk();
        //保留的字段
        String sinkColumns = tableProcess.getSinkColumns();
        //获取建表扩展
        String sinkExtend = tableProcess.getSinkExtend();

        //拼接状态中的key
        String key = sourceTable + ":" + operateType;

        //如果当前读取到的信息是维度配置 ----提前将维度表创建出来
        if(sinkType.equals(TableProcess.SINK_TYPE_HBASE)&&"insert".equals(operateType)){
            checkTable(sinkTable,sinkColumns,sinkPk,sinkExtend);
        }


        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDes);
        //将读取到的配置信息放到状态中
        broadcastState.put(key, tableProcess);
    }

    //创建维度表
    private void checkTable(String tableName, String fieldStr, String pk, String ext) {
        if(pk == null){
            pk = "id";
        }
        if(ext == null){
            ext = "";
        }

        StringBuilder createSql = new StringBuilder("create table if not exists "+ GmallConfig.HBASE_SCHEMA+"."+tableName+"(");
        //使用逗号对字段进行分隔
        String[] fieldArr = fieldStr.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String fieldName = fieldArr[i];
            if(i >= 1){
                createSql.append(",");
            }
            if(fieldName.equals(pk)){
                createSql.append(fieldName +" varchar primary key");
            }else{
                createSql.append(fieldName +" varchar ");
            }
        }
        createSql.append(")" + ext);

        System.out.println("phoenix的建表语句为:" + createSql);

        PreparedStatement ps = null;
        try {
            //创建数据库操作对象
            ps = conn.prepareStatement(createSql.toString());
            //执行SQL语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("在Phoenix中创建维度表失败~~~");
        }finally {
            //释放资源
            if(ps != null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}
